package org.apache.rocketmq.tools.topic;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.OptionGroup;
import org.apache.commons.cli.Options;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.AttributeParser;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.tools.command.CommandUtil;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommand;
import org.apache.rocketmq.tools.command.SubCommandException;

import java.util.Map;
import java.util.Set;

public class UpdateTopicSubCommand implements SubCommand {

    //返回该命令对象名称的方法
    @Override
    public String commandName() {
        return "updateTopic";
    }

    //返回该命令对象描述信息的方法
    @Override
    public String commandDesc() {
        return "Update or create topic.";
    }
    //执行命令的方法

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/15
     * @方法描述：把命令行中的参数封装到Options对象中的方法
     */
    @Override
    public Options buildCommandlineOptions(Options options) {
        //这个是我在测试类中设置的命令行参数信息
        //updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster
        OptionGroup optionGroup = new OptionGroup();
        //封装Broker地址命令参数
        Option opt = new Option("b", "brokerAddr", true, "create topic to which broker");
        optionGroup.addOption(opt);
        //封装集群名称命令参数
        opt = new Option("c", "clusterName", true, "create topic to which cluster");
        optionGroup.addOption(opt);
        //上面两个opt对象都设置到了一开始创建的OptionGroup对象中
        //这时候又把OptionGroup对象设置到了Options对象中
        //这个操作有一层含义，那就是如果用户在命令行中同时输入了-b和-c参数，那么就会报以下错误
        //The option 'c' was specified but an option from this group has already been selected: 'b'
        //原因很简单，-b 命令指定的是要创建主题的Broker节点的网络地址，也就是说在该节点上创建主题，而-c命令指定的是要创建主题的集群名称，也就是说在该集群中的所有主节点上创建主题。
        //一旦设置了其中一个命令，就不能再设置另一个命令，所以这里使用了OptionGroup对象，把这两个命令放到了一个组中，这样就可以保证只能设置其中一个命令。
        optionGroup.setRequired(true);
        options.addOptionGroup(optionGroup);

        //封装要创建的主题的属性命令参数，这里的属性就可以用来配置主题消息类型，可以把该主题定义为一个顺序消息主题
        //但现在我们还没有实现顺序消息功能，如果大家感兴趣，可以直接去MQ的官方文档中看看顺序消息的文档，里面介绍了如何使用当前命令对象创建顺序消息主题
        opt = new Option("a", "attributes", true, "attribute(+a=b,+c=d,-e)");
        opt.setRequired(false);
        options.addOption(opt);

        //封装要创建的主题的名称命令参数
        opt = new Option("t", "topic", true, "topic name");
        opt.setRequired(true);
        options.addOption(opt);

        //封装要创建的主题的读队列数命令参数
        opt = new Option("r", "readQueueNums", true, "set read queue nums");
        opt.setRequired(false);
        options.addOption(opt);

        //封装要创建的主题的写队列数命令参数
        opt = new Option("w", "writeQueueNums", true, "set write queue nums");
        opt.setRequired(false);
        options.addOption(opt);

        //封装要创建的主题的权限命令参数，2可写，4可读，6可读可写
        opt = new Option("p", "perm", true, "set topic's permission(2|4|6), intro[2:W 4:R; 6:RW]");
        opt.setRequired(false);
        options.addOption(opt);

        //封装要创建的主题的顺序消息命令参数
        opt = new Option("o", "order", true, "set topic's order(true|false)");
        opt.setRequired(false);
        options.addOption(opt);

        return options;
    }

    //执行命令的方法
    @Override
    public void execute(final CommandLine commandLine, final Options options,
                        RPCHook rpcHook) throws SubCommandException {

        //默认的控制台管理工具对象
        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
        //设置控制台管理工具对象的名称为当前时间戳
        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
        try {

            TopicConfig topicConfig = new TopicConfig();
            //设置默认读队列数为8
            topicConfig.setReadQueueNums(8);
            //设置默认写队列数为8
            topicConfig.setWriteQueueNums(8);
            //设置要创建的主题的名称
            topicConfig.setTopicName(commandLine.getOptionValue('t').trim());
            //设置主题的扩展属性信息
            if (commandLine.hasOption('a')) {
                //得到命令行参数中的属性信息
                String attributesModification = commandLine.getOptionValue('a').trim();
                //解析属性信息
                Map<String, String> attributes = AttributeParser.parseToMap(attributesModification);
                //设置主题的扩展属性信息
                topicConfig.setAttributes(attributes);
            }

            //设置主题的读队列数
            if (commandLine.hasOption('r')) {
                topicConfig.setReadQueueNums(Integer.parseInt(commandLine.getOptionValue('r').trim()));
            }

            //设置主题的写队列数
            if (commandLine.hasOption('w')) {
                topicConfig.setWriteQueueNums(Integer.parseInt(commandLine.getOptionValue('w').trim()));
            }

            //设置主题的权限
            if (commandLine.hasOption('p')) {
                topicConfig.setPerm(Integer.parseInt(commandLine.getOptionValue('p').trim()));
            }

            //设置主题是否支持顺序消息的标志
            boolean isOrder = false;
            if (commandLine.hasOption('o')) {
                isOrder = Boolean.parseBoolean(commandLine.getOptionValue('o').trim());
            }
            topicConfig.setOrder(isOrder);
            if (commandLine.hasOption('b')) {
                String addr = commandLine.getOptionValue('b').trim();
                defaultMQAdminExt.start();
                defaultMQAdminExt.createAndUpdateTopicConfig(addr, topicConfig);
                //判断是否支持顺序消息，以下操作暂时还用不上，因为我们还没有实现顺序消息功能
                //大家也可以暂时忽略
                if (isOrder) {
                    //如果支持则根据节点网络地址获取指定Broker节点的名称
                    String brokerName = CommandUtil.fetchBrokerNameByAddr(defaultMQAdminExt, addr);
                    //定义顺序消息配置信息
                    String orderConf = brokerName + ":" + topicConfig.getWriteQueueNums();
                    //在这里执行创建或者更新顺序消息配置信息的操作
                    defaultMQAdminExt.createOrUpdateOrderConf(topicConfig.getTopicName(), orderConf, false);
                    System.out.printf("%s%n", String.format("set broker orderConf. isOrder=%s, orderConf=[%s]", isOrder, orderConf.toString()));
                }
                System.out.printf("create topic to %s success.%n", addr);
                System.out.printf("%s%n", topicConfig);
                return;

            }else if (commandLine.hasOption('c')){
                defaultMQAdminExt.start();
                String clusterName = commandLine.getOptionValue('c');
                Set<String> brokerMasterAddrs = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
                for (String addr : brokerMasterAddrs) {
                    defaultMQAdminExt.createAndUpdateTopicConfig(addr, topicConfig);
                    System.out.printf("create topic to %s success.%n", addr);
                }
                //以下操作暂时还用不上，因为我们还没有实现顺序消息功能，大家也可以暂时忽略
                if (isOrder) {
                    Set<String> brokerNameSet = CommandUtil.fetchBrokerNameByClusterName(defaultMQAdminExt, clusterName);
                    StringBuilder orderConf = new StringBuilder();
                    String splitor = "";
                    for (String s : brokerNameSet) {
                        orderConf.append(splitor).append(s).append(":")
                                .append(topicConfig.getWriteQueueNums());
                        splitor = ";";
                    }
                    defaultMQAdminExt.createOrUpdateOrderConf(topicConfig.getTopicName(), orderConf.toString(), true);
                    System.out.printf("set cluster orderConf. isOrder=%s, orderConf=[%s]%n", isOrder, orderConf);
                }
                System.out.printf("%s%n", topicConfig);
                return;
            }
            //如果没有找到Broker节点或者集群名称，那么就打印命令行帮助信息
            ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
        } catch (Exception e) {
            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
        } finally {
            //程序执行完毕，关闭控制台程序
            defaultMQAdminExt.shutdown();
        }
    }


}
